package cn.jly.flink.distributed_cache;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.File;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * 分布式缓存在我们实际生产环境中最广泛的一个应用，就是在进行表与表 Join 操作时，如果一个表很大，另一个表很小，
 * 那么我们就可以把较小的表进行缓存，在每个 TaskManager 都保存一份，然后进行 Join 操作。
 *
 * 这四种级别的配置生效优先级如下：算子级别 > 执行环境级别 > 提交任务级别 > 系统配置级别
 *
 * @author lanyangji
 * @create 2020-09-02 9:41
 */
@Slf4j
public class DistributedCacheApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 需要注意的是，在实际生产环境中由于每个任务的负载和资源消耗不一样，我们推荐在代码中指定每个任务的重试机制和重启策略
        // 固定延迟失败重启策略模式，参数：重试次数，每次重试间隔
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3,
                Time.seconds(5)
        ));

        // 失败率重启策略模式
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, // 每个时间间隔的最大故障次数
//                Time.of(5, TimeUnit.MINUTES), // 测量故障率的时间间隔
//                Time.of(5, TimeUnit.SECONDS) //  每次任务失败时间间隔
//        ));

        // 获取缓存未见
        URL resource = DistributedCacheApp.class.getClassLoader().getResource("cache.txt");
        // 注册缓存文件
        env.registerCachedFile(resource.getPath(), "cacheFile");

        DataStreamSource<String> data = env.fromElements("tom", "lily", "james");
        data.map(new RichMapFunction<String, String>() {
            List<String> fileContent = new ArrayList<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                // 获取并使用缓存文件
                File cacheFile = getRuntimeContext().getDistributedCache().getFile("cacheFile");
                List<String> lines = FileUtils.readLines(cacheFile);
                fileContent.addAll(lines);
            }

            @Override
            public String map(String s) throws Exception {
                // 使用取出的分布式文件的内容
                int index = new Random().nextInt(fileContent.size());
                log.info("使用分布式缓存文件中的内容： {}-{}", fileContent.get(index), s);
                return fileContent.get(index) + "_" + s;
            }
        })
                .printToErr();

        env.execute("DistributedCacheApp");
    }
}
